package cn.getech.data.development.bean

import cn.getech.data.development.bean.dag.{FlinkStreamFieldObj, FlinkStreamTypeObj}
import cn.getech.data.development.bean.jdbc.JDBCSqlParserResourceBean
import cn.getech.data.development.enums.DAGType
import com.alibaba.fastjson.JSON

/**
 *
 */
/**
 * flink dag&sql model
 * @param flinkJson json文件路径
 * @param isDag dag:true; sql: false
 */
class FlinkStreamingObj(flinkJson: String, isDag: Boolean) extends FlinkStreamJsonObj {
  val obj = load(flinkJson)
  var sources: String = obj.getString("flinkSources")
  val isDebug:String = obj.getString("isDebug")
  var sqls: String = obj.getString("flinkSqls")
  var sinks: String = obj.getString("flinkSinks")
  var jobName: String = obj.getString("flinkJobName")
  var jdbcSinks = FlinkStreamJDBCObj.stringToJsonArrayParse(obj.getString("resource"))
  var typeObjs: Array[FlinkStreamTypeObj] = null

  // 校验
  var isSink = (sinks != null && !sinks.isEmpty)
  var isJdbcSink = (jdbcSinks != null && !jdbcSinks.isEmpty)

  // 只有dag需要调用这个方法
  override def jsonParse: Unit = {
    val objs = JSON.parseArray(sqls)
    typeObjs = objs.toArray.map(x => {
      val obj = JSON.parseObject(x.toString)
      val typeObj = new FlinkStreamTypeObj()
      typeObj.oper_type = DAGType.withName(obj.getOrDefault("oper_type", "").toString)
      typeObj.tableName = obj.getOrDefault("tableName", "").toString
      typeObj.outputTableName = obj.getOrDefault("outputTableName", "").toString
      typeObj.insertName = obj.getOrDefault("insertName", "").toString
      typeObj.sinkTableName = obj.getOrDefault("sinkTableName", "").toString

      val array = obj.getJSONArray("fields")
      typeObj.fields = array.toArray.map(x => {
        val fieldObj = FlinkStreamFieldObj
        val field = JSON.parseObject(x.toString)
        fieldObj.apply(field.getOrDefault("field_name", "").toString,
          field.getOrDefault("alias", "").toString,
          field.getOrDefault("class_type", "").toString,
          field.getOrDefault("function", "").toString)
      })

      typeObj.fieldText = obj.getOrDefault("fieldText", "").toString

      typeObj.groupName = obj.getOrDefault("groupName", "").toString
      typeObj.isWindow = obj.getOrDefault("isWindow", "false").toString.toBoolean
      if (typeObj.isWindow) {
        typeObj.windowType = obj.getOrDefault("windowType", "").toString
        typeObj.windowName = obj.getOrDefault("windowName", "").toString
        typeObj.interval = obj.getOrDefault("interval", "0").toString.toLong
        typeObj.unit = obj.getOrDefault("unit", "").toString
      }

      typeObj.partition = obj.getOrDefault("partition", "").toString
      typeObj.order = obj.getOrDefault("order", "").toString
      typeObj.alias_partition = obj.getOrDefault("alias_partition", "").toString
      typeObj
    })

  }
}
